package org.fhm.zdte.common.service.message;

import org.fhm.substrate.annotation.Setup;
import org.fhm.substrate.manager.Bootstrap;
import org.fhm.zdte.common.ability.IGetIdentification;
import org.fhm.zdte.common.ability.IGetServerAddress;
import org.fhm.zdte.common.ability.IGetTargetAddress;
import org.fhm.zdte.common.ability.IRequestTimeout;
import org.fhm.zdte.common.config.ProtocolConfiguration;
import org.fhm.zdte.common.constant.CommonConstant;
import org.fhm.zdte.common.constant.DataTransform;
import org.fhm.zdte.common.constant.protocol.ContentType;
import org.fhm.zdte.common.constant.protocol.PacketType;
import org.fhm.zdte.common.constant.subcmd.RequestCmd;
import org.fhm.zdte.common.exception.ProtocolException;
import org.fhm.zdte.common.pojo.Address;
import org.fhm.zdte.common.pojo.Request;
import org.fhm.zdte.common.pool.ThreadPoolFactory;
import org.fhm.zdte.common.protocol.AbstractMessageStandard;
import org.fhm.zdte.common.protocol.AbstractMessageWrappers;
import org.fhm.zdte.common.standard.IChannelDispatcher;
import org.fhm.zdte.common.standard.ISendMessageManager;
import org.fhm.zdte.common.task.SendDealTask;
import org.fhm.zdte.common.util.CommonUtil;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * The abstract implementation for {@link ISendMessageManager}.
 *
 * @author 谭波
 * @since 2023/11/15
 */
public abstract class AbstractSendMessageManager implements ISendMessageManager, IRequestTimeout {

    private static final Map<String, Map<Integer, AbstractMessageStandard>> sendMessagePool = new ConcurrentHashMap<>();

    private static final Map<String, Set<String>> userMessagePool = new ConcurrentHashMap<>();

    private static final Map<String, ScheduledFuture<?>> scheduleMessageTaskPool = new ConcurrentHashMap<>();

    private static final Map<String, Address> messageAddressBufferPool = new ConcurrentHashMap<>();

    private static final Map<String, byte[]> messageBufferPool = new ConcurrentHashMap<>();
    protected final ContentType contentType;
    private final ThreadLocal<Random> randomThreadLocal = new ThreadLocal<>();
    protected int lengthOfPacketWithContent;
    protected int resendPacketCount;
    protected int firstSendPacketCount;
    @Setup("->protocol.name")
    protected AbstractMessageWrappers abstractMessageWrappers;

    @Setup
    protected IGetIdentification iGetIdentification;

    @Setup
    protected IChannelDispatcher iChannelDispatcher;

    @Setup
    protected ThreadPoolFactory threadPoolFactory;

    @Setup
    protected CommonUtil commonUtil;
    @Setup
    protected SendDealTask sendDealTask;
    @Setup
    private IGetTargetAddress iGetTargetAddress;
    @Setup
    private IGetServerAddress iGetServerAddress;
    @Setup
    private ProtocolConfiguration protocolConfiguration;

    public AbstractSendMessageManager(ContentType contentType) {
        this.contentType = contentType;
    }

    @Override
    public void initial(Request bean, String messageNumber, String target) {
        Address targetAddress = iGetTargetAddress.getTargetAddress(target);
        if (Objects.isNull(targetAddress))
            targetAddress = iGetServerAddress.getServerAddress();
        if (Objects.isNull(targetAddress))
            throw new ProtocolException("the sending destination has been disconnected");
        messageAddressBufferPool.put(messageNumber, targetAddress);
        sendMessagePool.computeIfAbsent(messageNumber, k -> new ConcurrentHashMap<>());
        userMessagePool.computeIfAbsent(target, k -> new CopyOnWriteArraySet<>()).add(messageNumber);
        byte[] bytes = convert(bean);
        messageBufferPool.put(messageNumber, bytes);
        int realLen = bytes.length;
        int messageLength = realLen < lengthOfPacketWithContent || lengthOfPacketWithContent == 0
                ? 1 : (realLen / lengthOfPacketWithContent + 1);
        final Address finalTargetAddress = targetAddress;
        scheduleMessageTaskPool.put(
                messageNumber,
                threadPoolFactory.submitPeriodicTask(
                        () -> iChannelDispatcher
                                .obtainChannelManager(finalTargetAddress)
                                .sendMessage(abstractMessageWrappers.getMessageByPacketType(
                                        PacketType.REQUEST,
                                        RequestCmd.INITIAL,
                                        iGetIdentification.getIdentification(),
                                        target,
                                        messageNumber,
                                        messageLength,
                                        null,
                                        null,
                                        null
                                ).setContentType(contentType), finalTargetAddress),
                        1,
                        TimeUnit.SECONDS
                )
        );
    }

    /**
     * Convert the request object into a suitable byte array.
     *
     * @param request Request object.
     * @return Byte array.
     */
    protected abstract byte[] convert(Request request);

    @Override
    public void cancelInitialTaskAndSend(String messageNumber, Address address) {
        if (Objects.nonNull(messageBufferPool.get(messageNumber))) {
            synchronized (this) {
                byte[] bytes;
                if (Objects.nonNull((bytes = messageBufferPool.remove(messageNumber)))) {
                    commonUtil.cancelTask(scheduleMessageTaskPool.remove(messageNumber));
                    addSubMessageAndSend(bytes, messageNumber, messageAddressBufferPool.remove(messageNumber));
                }
            }
        }
    }

    private void addSubMessageAndSend(
            byte[] bytes,
            String messageNumber,
            Address address
    ) {
        int realLen = bytes.length;
        if (realLen == 0 || realLen <= lengthOfPacketWithContent) {
            threadPoolFactory.submitTask(
                    ThreadPoolFactory.TRANSFORM_THREAD_POOL,
                    () -> recordAndSendMessage(warpSubMessage(
                            warpMessageStream(bytes, 0, realLen),
                            1,
                            0,
                            messageNumber
                    ).setContentType(contentType), address)
            );
        } else {
            int finalIdx;
            int messageLength = (
                    (finalIdx = realLen % lengthOfPacketWithContent) == 0 ?
                            realLen / lengthOfPacketWithContent : realLen / lengthOfPacketWithContent + 1
            );
            int messageSerial = 0;
            int lastIdx = realLen - finalIdx;
            for (
                    int start = 0; start < lastIdx;
                    start += lengthOfPacketWithContent, messageSerial++
            ) {
                int finalMessageSerial = messageSerial;
                int finalStart = start;
                threadPoolFactory.submitTask(
                        ThreadPoolFactory.TRANSFORM_THREAD_POOL,
                        () -> recordAndSendMessage(warpSubMessage(
                                warpMessageStream(bytes, finalStart, lengthOfPacketWithContent),
                                messageLength,
                                finalMessageSerial,
                                messageNumber
                        ).setContentType(contentType), address)
                );
            }
            int finalMessageSerial = messageSerial;
            if (finalIdx > 0) {
                threadPoolFactory.submitTask(
                        ThreadPoolFactory.TRANSFORM_THREAD_POOL,
                        () -> recordAndSendMessage(warpSubMessage(
                                warpMessageStream(bytes, lastIdx, finalIdx),
                                messageLength,
                                finalMessageSerial,
                                messageNumber
                        ).setContentType(contentType), address)
                );
            }
            refreshServerOnlineTime();
        }
    }

    private byte[] warpMessageStream(byte[] bytes, int start, int pos) {
        byte[] chunk;
        System.arraycopy(bytes, start, (chunk = new byte[pos]), 0, pos);
        return chunk;
    }

    private AbstractMessageStandard warpSubMessage(
            byte[] messageStream, int messageLen, int messageSerial, String messageNumber
    ) {
        return abstractMessageWrappers.getMessageWithContent(
                PacketType.REQUEST,
                RequestCmd.SEND,
                iGetIdentification.getIdentification(),
                messageNumber,
                messageLen,
                messageSerial,
                messageStream
        );
    }

    private void recordAndSendMessage(AbstractMessageStandard subMessage, Address address) {
        Integer serialNumber = subMessage.getSerialNumber();
        sendMessagePool.get(subMessage.getMessageNumber())
                .put(serialNumber, subMessage.setContentType(contentType));
        if (firstSendPacketCount == 0 || serialNumber < firstSendPacketCount)
            iChannelDispatcher.obtainChannelManager(address)
                    .sendMessage(subMessage, address);
    }

    @Override
    public void resendResponse(AbstractMessageStandard message, Address address) {
        String messageNumber = message.getMessageNumber();
        String content = new String(message.getContent(), Bootstrap.charset);
        if (content.equals(CommonConstant.MESSAGE_COMPLETE_MARK.getName())) {
            sendCompleteAckClearMessage(messageNumber, address);
        } else {
            String[] successfulMessageNos
                    = content.split(CommonConstant.PACKAGE_SEQUENCE_NUMBER_SEPARATOR.getName());
            Map<Integer, AbstractMessageStandard> messages = sendMessagePool.get(messageNumber);
            if (Objects.nonNull(messages)) {
                synchronized (messages) {
                    if (Objects.nonNull(sendMessagePool.get(messageNumber))) {
                        for (String no : successfulMessageNos)
                            messages.remove(Integer.parseInt(no));
                        if (messages.values().stream().allMatch(Objects::isNull)) {
                            iChannelDispatcher.obtainChannelManager(address)
                                    .sendMessage(abstractMessageWrappers.getMessageWithContent(
                                            PacketType.REQUEST,
                                            RequestCmd.SEND,
                                            iGetIdentification.getIdentification(),
                                            messageNumber,
                                            0,
                                            0,
                                            CommonConstant
                                                    .MESSAGE_COMPLETE_MARK
                                                    .getName()
                                                    .getBytes(Bootstrap.charset)
                                    ).setContentType(contentType), address);
                        } else {
                            resend(messageNumber, address);
                            refreshServerOnlineTime();
                        }
                    }
                }
            } else {
                iChannelDispatcher.obtainChannelManager(address)
                        .sendMessage(abstractMessageWrappers.getSuccessfulAckMessage(
                                PacketType.REQUEST,
                                RequestCmd.SEND,
                                iGetIdentification.getIdentification(),
                                messageNumber
                        ).setContentType(contentType), address);
            }
        }
    }

    private void resend(String messageNumber, Address address) {
        Random random = commonUtil.obtainObjOfThreadLocal(randomThreadLocal, Random::new);
        Map<Integer, AbstractMessageStandard> messages = sendMessagePool.get(messageNumber);
        List<Integer> keys = new ArrayList<>(messages.keySet());
        if (resendPacketCount != 0)
            for (int i = 0; i < resendPacketCount && !keys.isEmpty(); i++) {
                AbstractMessageStandard message = messages.get(keys.remove(random.nextInt(keys.size())));
                if (Objects.nonNull(message)) {
                    threadPoolFactory.submitTask(
                            ThreadPoolFactory.TRANSFORM_THREAD_POOL,
                            () -> iChannelDispatcher.obtainChannelManager(address)
                                    .sendMessage(message, address)

                    );
                }
            }
    }

    private void refreshServerOnlineTime() {
        sendDealTask.getTaskAndSubmit(
                abstractMessageWrappers
                        .getAckMessageByPacketType(
                                PacketType.ONLINE_CHECK, null,
                                iGetIdentification.getIdentification(), "",
                                null, null,
                                ""),
                iGetServerAddress.getServerAddress()
        );
    }

    private void sendCompleteAckClearMessage(String messageNumber, Address address) {
        iChannelDispatcher.obtainChannelManager(address)
                .sendMessage(abstractMessageWrappers.getSuccessfulAckMessage(
                        PacketType.REQUEST,
                        RequestCmd.SEND,
                        iGetIdentification.getIdentification(),
                        messageNumber
                ).setContentType(contentType), address);
        sendMessagePool.remove(messageNumber);
        for (Set<String> messagePool : userMessagePool.values()) {
            if (messagePool.contains(messageNumber)) {
                messagePool.remove(messageNumber);
                break;
            }
        }
    }


    @Override
    public void clearMessageBuffer(String from) {
        Set<String> messageNumbers = userMessagePool.get(from);
        if (Objects.nonNull(messageNumbers)) {
            synchronized (messageNumbers) {
                if (Objects.nonNull(userMessagePool.get(from))) {
                    messageNumbers.forEach(
                            messageNumber -> {
                                commonUtil.cancelTask(scheduleMessageTaskPool.remove(messageNumber));
                                sendMessagePool.remove(messageNumber);
                                messageAddressBufferPool.remove(messageNumber);
                                messageBufferPool.remove(messageNumber);
                            }
                    );
                    userMessagePool.remove(from);
                }
            }
        }
    }

    /**
     * Configure the length that Packet-Data with content.
     */
    protected void setLengthOfPacketWithContent() {
        lengthOfPacketWithContent = protocolConfiguration.getPacketLength() -
                abstractMessageWrappers.getLengthOfMessageWithContent(commonUtil.getIdentification(null));
    }

    /**
     * Set the number of resent packages.
     */
    protected void setResendPacketCount() {
        resendPacketCount = DataTransform.NUMBER_OF_FLOWS * 3;
    }

    /**
     * Set the number of first send packages.
     */
    protected void setFirstSendPacketCount() {
        firstSendPacketCount = DataTransform.NUMBER_OF_FLOWS * 12;
    }

    @Override
    public void connectTimeoutRun(String form, String messageNumber) {
        userMessagePool.get(form).remove(messageNumber);
        commonUtil.cancelTask(scheduleMessageTaskPool.remove(messageNumber));
        sendMessagePool.remove(messageNumber);
        messageAddressBufferPool.remove(messageNumber);
        messageBufferPool.remove(messageNumber);
    }
}
